/*
 * ===================================================================
 *  
 *      Filename:  log_client_policy.c
 *
 *   Description:  This file is used to maitain the policy file.
 *                 Update/Remove/Create/RDWR will be supplied.
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *          
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <signal.h>
#include <pthread.h>
#include <linux/unistd.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include "lib/util.h"
#include "lib/list.h"
#include "log_client_api.h"

int global_policy_version = 0;

int get_policy_version(char* pathname)
{
    struct stat file_status;
    char   *buff,*tmp_buff,*tmp_msg_buff;
    int    icounter,fd;

    if(stat(pathname,&file_status))
    {
        return 0;
    }
    buff = (char *)malloc(file_status.st_size);
    if(!buff)
    {
        return 0;
    }

    /*during the update stage, new connected client will NOT be notified!!!!!!*/
    if(-1 == (fd = open(pathname,O_RDONLY)))
    {
        free(buff);
        return 0;
    }
    tmp_buff = buff;
    while((icounter = read(fd,tmp_buff,SSIZE_MAX))>0)
    {
        tmp_buff = tmp_buff + icounter;
    }
    
    /*close the policy file*/
    close(fd);

    /*read the version from the file*/
    /*now we get all of the msg in the buff*/
    tmp_msg_buff = strstr(buff,"version");
    if(tmp_msg_buff)
    {
        tmp_msg_buff = tmp_msg_buff + 7;//strlen("version")
        tmp_msg_buff = strstr(tmp_msg_buff,"version");
        if(tmp_msg_buff)
        {
            while(*tmp_msg_buff != '=')
            tmp_msg_buff++;
            tmp_msg_buff++;
            return atoi(tmp_msg_buff);
        }
        else
            return 0;

    }
    else
        return 0;
}

int parse_color_msg(unsigned char* buff,int buff_len, int policy_fd,struct sockaddr_in* addr,socklen_t addr_len)
{
    int color_request;
    int iret;
    color_request = buff[1];
    iret = set_color_status(color_request,buff);
    if(iret)
    {
        /*failed to set the operation,reject*/
        buff[1] = REJECT_COLOR_REQUEST;
    }
    else
    {
        if(color_request == ENABLE_COLOR_REQUEST)
            buff[1] = ENABLE_COLOR_REPLY;
        else if(color_request == DISABLE_COLOR_REQUEST)
            buff[1] = DISABLE_COLOR_REPLY;
        else
            buff[1] = REJECT_COLOR_REQUEST;
    }
    sendto(policy_fd,buff,2,0,(struct sockaddr *)addr,(socklen_t)addr_len);
    return NOERR;
}

void rename_policy_file()
{
    remove(POLICY_FILE_PATH);
}

int parse_log_level_msg(unsigned char* buff,int buff_len, int policy_fd,struct sockaddr_in* addr,socklen_t addr_len)
{
	return set_daemon_log_level(buff[1]);
}

int parse_loadbalance_msg(unsigned char* buff,int buff_len, int policy_fd,struct sockaddr_in* addr,socklen_t addr_len)
{
    update_destination_addr(SEARCH_ALL);
    return NOERR;
}

int tmp_policy_fd = INV_HANDLER;
int parse_policy_msg(unsigned char* buff,int buff_len, int policy_fd,struct sockaddr_in* addr,socklen_t addr_len)
{
    int policy_request;
    static int update_flag = 0; /*indicate the update is starting*/
    unsigned int pre_addr;
    int iret = 0;

    policy_request = buff[1];
    /*update request*/
    if(policy_request == UPDATE_POLICY_REQUEST)
    {   
        if(update_flag)
        {
            log_doctor(doctor_level,"in progress,reject the new request\n");
            goto REJECT_REQUEST;
        }
        else
        {
            if(*((int*)(buff+2)) <= get_policy_version(POLICY_FILE_PATH))
            {
                log_doctor(doctor_level,"the policy version is %d,less or equal then the local policy\n",*((int*)(buff+2)));
                goto REJECT_REQUEST;
            }
            /*rename the policy file*/
            rename_policy_file();
            tmp_policy_fd = open(POLICY_FILE_PATH,O_APPEND|O_CREAT|O_RDWR,00666);
            if(tmp_policy_fd == -1)
            {
                log_doctor(doctor_level,"Failed to open the policy file\n");
                goto REJECT_REQUEST;
            }
            else
            {
                buff[1] = UPDATE_POLICY_ACCEPT;
                sendto(policy_fd,buff,2,0,(struct sockaddr *)addr,addr_len);
                update_flag = 1;
                return NOERR;
            }
        }
REJECT_REQUEST:
        buff[1] = UPDATE_POLICY_REJECT;
        iret = sendto(policy_fd,buff,2,0,(struct sockaddr *)addr,addr_len);
        if(iret == -1)
        {
            perror("error to send the reply");
        }
        return NOERR;
    }
    /*finish the update,return*/
    else if(policy_request == UPDATE_POLICY_FINISH)
    {
        close(tmp_policy_fd);
        tmp_policy_fd = INV_HANDLER;
        /*hold the last time srv addr*/
        pre_addr = get_destination();
        /*choose the destination srv again*/
        update_destination_addr(SEARCH_ALL);
        if(pre_addr != get_destination())
        {
            /*the srv address changed,recreate the socket*/
            reinit_protocol_tcp(get_destination());
            reinit_protocol_udp(get_destination());
        }
        /*flush the local backup file*/
        flush_backup_to_srv_with_tcp();        
        update_flag = 0;
        return NOERR;
    }
    /*update continue*/
    else if(policy_request == UPDATE_POLICY_DATA)
    {
        if(!update_flag)
            return FUDPY;
        /*write the data to the policy file*/
        if(tmp_policy_fd == INV_HANDLER)
            return FUDPY;
        write(tmp_policy_fd,buff+2,buff_len-2);
        return NOERR;
    }
    /*I also don't know*/
    else
    {
        log_doctor(doctor_level,"bug,return to the policy daemon\n");
        return FUDPY;
    }
    return NOERR;
}

int  parse_command(unsigned char* buff,int recv_len,int policy_fd,struct sockaddr_in* addr, socklen_t addr_len)
{
    if((*buff)>>4 == CS_PROTOCOL_COLOR)
        return parse_color_msg(buff,recv_len,policy_fd,addr,addr_len);
    else if((*buff)>>4 == CS_PROTOCOL_POLICY)
        return parse_policy_msg(buff,recv_len,policy_fd,addr,addr_len);
    else if((*buff)>>4 == CS_PROTOCOL_LOG)
		return parse_log_level_msg(buff,recv_len,policy_fd,addr,addr_len);
    else if((*buff)>>4 == CS_PROTOCOL_LB)
		return parse_loadbalance_msg(buff,recv_len,policy_fd,addr,addr_len);
    return NOERR;
}

void* color_daemon(void* argv)
{
    int color_fd = (long)argv;
    int recv_len;
    socklen_t remote_client_len;
    unsigned char   buff[MAX_LOG_MSG_SIZE];
    struct sockaddr_in remote_client;

    while(1)
    {
        recv_len = recvfrom(color_fd,buff,MAX_LOG_MSG_SIZE,0,(struct sockaddr*)&remote_client,(socklen_t *)&remote_client_len);
        if(recv_len > 0)
        {
            /*add the terminal flag to the end of the file*/
            buff[recv_len] = '\0';
            /*get color msg from client*/
            parse_command(buff,recv_len,color_fd,&remote_client,remote_client_len );
        }
        else
        {
            perror("error to recv from the policy socket");
            continue;
        }
    }
    return NULL;
}

#if MULCAST
struct ip_mreq ipmr;
int mulcast_sock_id;
void* policy_daemon(void *argv)
{
    struct sockaddr_in addr, sender;
    unsigned char buf[MAX_LOG_MSG_SIZE];
    socklen_t len;
    int ret;
    int count;

    /* Step 1: open a socket, and bind */
    if ((mulcast_sock_id = socket(AF_INET, SOCK_DGRAM, 0)) < 0) 
    {
        perror("socket error");
        exit(1);
    }

    memset((void*)&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port = htons(HPLOG_CLT_MULC);

    if (bind(mulcast_sock_id, (struct sockaddr *)&addr, sizeof(addr)) < 0) 
    {
        perror("bind error");
        exit(1);
    }   

    /* Step 2: fill in a struct ip_mreq */
    memset((void*)&ipmr, 0, sizeof(ipmr));
    ipmr.imr_multiaddr.s_addr = inet_addr(POLICY_MULT_GROUP); /* multicast group ip */
    ipmr.imr_interface.s_addr = htonl(INADDR_ANY);

    /* Step 3: call setsockopt with IP_ADD_MEMBERSHIP to support receiving multicast */
    if (setsockopt(mulcast_sock_id, IPPROTO_IP, IP_ADD_MEMBERSHIP, &ipmr, sizeof(ipmr)) < 0) 
    {
        perror("setsockopt:IP_ADD_MEMBERSHIP");
        exit(1);
    }

    /* Step 4: call recvfrom to receive multicast packets */
    len = sizeof(sender);
    count = 0;
    while (1) 
    {
        ret = recvfrom(mulcast_sock_id, buf, MAX_LOG_MSG_SIZE, 0, (struct sockaddr *)&sender, &len);
        buf[ret] = '\0';
        if (ret < 0) 
        {
            perror("recvfrom error");
            exit(1);
        }
        else
        {
            parse_command(buf,ret,mulcast_sock_id,&sender,len );
        }
    }

}
#endif

#if MULCAST
pthread_t policy_pid;
#endif
pthread_t color_pid;
pthread_t init_log_policy(int policy_fd)
{
    long local_policy_fd = policy_fd;
    if(pthread_create(&color_pid,NULL,color_daemon,(void*)local_policy_fd))
    {
        return INV_HANDLER;
    }
    #if MULCAST
    if(pthread_create(&policy_pid,NULL,policy_daemon,NULL))
    {
        return INV_HANDLER;
    }
    #endif
    return color_pid;
}

int finit_log_policy()
{
    #if MULCAST
    if (setsockopt(mulcast_sock_id, IPPROTO_IP, IP_DROP_MEMBERSHIP, &ipmr, sizeof(ipmr)) < 0) 
    {
        perror("setsockopt:IP_DROP_MEMBERSHIP");
        exit(1);
    }
    close(mulcast_sock_id);
    pthread_cancel(policy_pid);
    #endif
    pthread_cancel(color_pid);
    return NOERR;
}

